package org.bear.bundle.framework;

import java.io.File;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.BlobMessage;

/**
 * JMS消息生产者
 *
 */
public class JMSProducer implements ExceptionListener {

	public Properties properties = new Properties();

	// 设置连接的最大连接数
	public final static int DEFAULT_MAX_CONNECTIONS = 5;
	public final static int DEFAULT_MAXIMUM_ACTIVE_SESSION_PER_CONNECTION = 300;
	// 强制使用同步返回数据的格式
	private boolean useAsyncSendForJMS = DEFAULT_USE_ASYNC_SEND_FOR_JMS;
	public final static boolean DEFAULT_USE_ASYNC_SEND_FOR_JMS = true;
	// 是否持久化消息
	private boolean isPersistent = DEFAULT_IS_PERSISTENT;
	public final static boolean DEFAULT_IS_PERSISTENT = false;

	// 连接地址
	private String brokerUrl;

	private String username;

	private String password;

	private ActiveMQConnectionFactory actualConnectionFactory;

	public JMSProducer(String brokerUrl, String username, String password) {
		this(brokerUrl, username, password, DEFAULT_MAX_CONNECTIONS,
				DEFAULT_MAXIMUM_ACTIVE_SESSION_PER_CONNECTION,
				DEFAULT_USE_ASYNC_SEND_FOR_JMS, DEFAULT_IS_PERSISTENT);
	}

	public JMSProducer(String brokerUrl, String username, String password,
			int maxConnections, int maximumActiveSessionPerConnection,
			boolean useAsyncSendForJMS, boolean isPersistent) {
		this.useAsyncSendForJMS = useAsyncSendForJMS;
		this.isPersistent = isPersistent;
		this.brokerUrl = brokerUrl;
		this.username = username;
		this.password = password;
		init();
	}

	private void init() {
		// ActiveMQ的连接工厂
		actualConnectionFactory = new ActiveMQConnectionFactory(this.username,
				this.password, this.brokerUrl);
		actualConnectionFactory.setUseAsyncSend(this.useAsyncSendForJMS);
	}

	/**
	 * 执行发送消息的具体方法
	 * 
	 * @param queue
	 * @param file
	 */
	public void send(final String queue, final Map<String, Object> map,
			final Object object) {
		try {
			sendMsg(queue, map, object);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 真正的执行消息发送
	 * 
	 * @param queue
	 * @param file
	 * @throws Exception
	 */
	private void sendMsg(String queue, final Map<String, Object> map,
			final Object object) throws Exception {

		Connection connection = null;
		ActiveMQSession session = null;
		try {
			connection = this.actualConnectionFactory.createConnection();
			session = (ActiveMQSession) connection.createSession(Boolean.FALSE,
					Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue(queue);
			MessageProducer producer = session.createProducer(destination);
			producer.setDeliveryMode(this.isPersistent ? DeliveryMode.PERSISTENT
					: DeliveryMode.NON_PERSISTENT);

			Message message = null;
			if (object instanceof String) {
				message = getMessage(session, map, object.toString());
			} else {
				if (object == null) {
					message = getMessage(session, map);
				} else {
					message = getMessage(session, map, (File) object);
				}
			}

			producer.send(message);
		} finally {
			closeSession(session);
			closeConnection(connection);
		}
	}

	private Message getMessage(ActiveMQSession session,
			Map<String, Object> map, File file) throws JMSException {
		BlobMessage message = session.createBlobMessage(file);
		if (map != null && !map.isEmpty()) {
			Set<String> keys = map.keySet();
			for (String key : keys) {
				message.setObjectProperty(key, map.get(key));
			}
		}
		return message;
	}

	private Message getMessage(Session session, Map<String, Object> map)
			throws JMSException {
		MapMessage message = session.createMapMessage();
		if (map != null && !map.isEmpty()) {
			Set<String> keys = map.keySet();
			for (String key : keys) {
				message.setObject(key, map.get(key));
			}
		}
		return message;
	}

	private Message getMessage(Session session, Map<String, Object> map,
			String string) throws JMSException {
		TextMessage message = session.createTextMessage();
		if (map != null && !map.isEmpty()) {
			Set<String> keys = map.keySet();
			for (String key : keys) {
				message.setObjectProperty(key, map.get(key));
			}
		}
		message.setText(string);
		return message;
	}

	public void closeSession(Session session) {
		try {
			if (session != null) {
				session.close();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void closeConnection(Connection connection) {
		try {
			if (connection != null) {
				connection.close();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void onException(JMSException e) {
		e.printStackTrace();
	}

}
